package com.ibm.cps.message;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonParser;
import com.ibm.cps.message.MessageFactory.MessageType;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class RuleMetaMessage extends ProcessorMetaMessage implements
		KryoSerializable {
	public String rulename;
	public String rule;
	public String topicname;
	public boolean model;

	public RuleMetaMessage() {
	}

	public RuleMetaMessage(String tenantid, String ruleName, String ruleStr,
			String topicName, String isEvent, String ruleId,
			Set<String> parentProcessors) {
		this.tenantid = tenantid;
		this.rulename = ruleName;
		this.rule = ruleStr;
		this.topicname = topicName;
		this.model = isEvent.equals("true") ? true : false;
		this.processorid = ruleId;
		this.parentprocessors = parentProcessors;
		this.type = MessageType.ADD_RULE_TOPIC;
	}

	/**
	 * json format
	 * tenantid schemaid topicName rulename ruleStr
	 * @param ruleString
	 */
	public RuleMetaMessage(byte[] ruleByte) {
		try {

			String ruleString = new String(ruleByte);
			ObjectNode objectNode = (ObjectNode) JSONUtility
					.fromString(ruleString);
			tenantid = objectNode.get(MessageFields.TENANTID).asText();
			rulename = objectNode.get(MessageFields.RULENAME).asText();
			rule = objectNode.get(MessageFields.RULE).asText();
			topicname = objectNode.get(MessageFields.TOPICNAME).asText();
			model = objectNode.get(MessageFields.MODEL).asBoolean();
			processorid = objectNode.get(MessageFields.PROCESSORID).asText();
			ArrayNode array = (ArrayNode) objectNode
					.get(MessageFields.PARENTPROCESSORS);
			parentprocessors = new HashSet<String>(array.size());
			for (int i = 0; i < array.size(); i++) {
				parentprocessors.add(array.get(i).asText());
			}
			type = MessageType.ADD_RULE_TOPIC;
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	private void writeObject(ObjectOutputStream out) throws IOException {
		out.defaultWriteObject();
		out.writeObject(type);
		out.writeObject(tenantid);
	}

	private void readObject(ObjectInputStream in)
			throws ClassNotFoundException, IOException {
		in.defaultReadObject();
		type = (MessageType) in.readObject();
		tenantid = (String) in.readObject();
	}

	public String getRuleName() {
		return rulename;
	}

	public String getRule() {
		return rule;
	}

	public String getTopicName() {
		return topicname;
	}

	public boolean isEventType() {
		return model;
	}

	public String getRuleJsonStr() {
		Gson gson = new Gson();
		String json = gson.toJson(this);
		try {
			ObjectNode objectNode = JSONUtility.fromObjectString(json);
			objectNode.remove("type");
			objectNode.put("type", "rule");
			objectNode.remove("deletion");
			return objectNode.toString();
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(processorid);
		output.writeString(rulename);
		output.writeString(rule);
		output.writeString(tenantid);
		kryo.writeObject(output, MessageType.ADD_RULE_TOPIC);
		output.writeString(topicname);
		output.writeBoolean(model);
		kryo.writeObject(output, parentprocessors);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		processorid = input.readString();
		rulename = input.readString();
		rule = input.readString();
		tenantid = input.readString();
		type = kryo.readObject(input, MessageType.class);
		topicname = input.readString();
		model = input.readBoolean();
		// Spark Workaround here
		parentprocessors = kryo.readObject(input, HashSet.class);
	}

	/**
	 * n to 1 processor
	 */
	@Override
	public void updateProcessorMap(
			HashMap<String, Set<String>> parent2Successors) {
		for (String parent : parentprocessors) {
			if (parent2Successors.get(parent + MessageFields.PROCESSOR_SUFFIX) == null) {
				parent2Successors.put(parent + MessageFields.PROCESSOR_SUFFIX,
						new HashSet<String>());
			}
			parent2Successors.get(parent + MessageFields.PROCESSOR_SUFFIX).add(
					processorid);
		}
	}

	public String getRuleString() {
		return this.rule;
	}

	public String getName() {
		return this.rulename;
	}

	public String getUserid() {
		return tenantid;
	}

	public boolean getModel() {
		return model;
	}

	public void setModel(boolean model) {
		this.model = model;
	}

	public static String getCollectionRuleRecord(
			Collection<RuleMetaMessage> rules) {
		JsonArray arrayNode = new JsonArray();
		JsonParser parser = new JsonParser();
		for (RuleMetaMessage rule : rules) {
			arrayNode.add(parser.parse(rule.getRuleJsonStr()));
		}
		return arrayNode.toString();
	}
}
